package com.navi.job;

import com.navi.service.impl.RabbitService;
import com.rabbitmq.http.client.domain.QueueInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.navi.constant.GenericDef.*;

/**
 * Created by lin.xin on 2017/10/27.
 */
@Service
@Slf4j
public class HmsMainLoaderJob {

    @Value("${spring.rabbitmq.queues}")
    private String queues;

    @Value("${hms.mode}")
    private String mode;
    @Value("${fetch_cnt}")
    private Integer fetchCnt;
    @Value("${app.limit_time}")
    private String limit_time;
    @Value("${app.deleteInsert}")
    private String deleteInsert;

    @Autowired
    private HmsLoaderService hmsLoaderService2;

    @Autowired
    private RabbitService rabbitService;

    @Scheduled(fixedDelayString = "${hms.fixedDelay}")
    public void loaderHistoryProc() {
        log.info("HMS system start ......");
        log.info("HMS start param :  " +
                        "queues [{}], mode [{}], fetchCnt [{}], deleteInsert [{}], limit_time [{}], ",
                queues, mode, fetchCnt, deleteInsert, limit_time);

        String processQueues = queues;
        int realFetchCount = fetchCnt;
        if (StringUtils.equalsIgnoreCase(mode, MODE_ERROR)) {
            processQueues = HMS_ERROR_Q;
        }

        if (HMS_ERROR_Q_BY_TABLE.equals(queues)) {
            // 处理所有error Q  除了 HMS_ERROR2
            processQueues = rabbitService.getQueueInfo().stream()
                    .filter(q -> q.getName().startsWith(HMS_ERROR_Q_BY_TABLE_PREFIX))
                    .map(QueueInfo::getName)
                    .collect(Collectors.joining(","));
            if (StringUtils.isNotEmpty(processQueues)) {
                processQueues = processQueues.concat(",").concat(HMS_ERROR_Q);
            } else {
                processQueues = HMS_ERROR_Q;
            }

            //TODO 如果 fetchCnt > 2  & 消息（被压缩过，每个消息中多笔数据）数量 = 2, 一旦压缩消息中有一笔数据有问题，则整笔消息均处理不掉
            // 这里其实可以拿到队列深度的， 可以判断当深度 eg:  = 2的时候fetchCnt传入1 (这个公式不严谨， 要思考下，或许要结合 compressJsonObjectMessage处理)
            // 不能直接修改fetchCnt，参考 processQueues 声明新变量 & 赋值  => realFetchCount

        }
        log.info("Process Q {}.", processQueues);

        Stream.of(processQueues.split(",")).forEach(queueName -> {
            log.info("Process Q {}, count {}.", queueName, realFetchCount);
            hmsLoaderService2.subMainProc(queueName, realFetchCount, limit_time, deleteInsert);
        });
    }
}
